package MQ.Exchange.Topic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

@SuppressWarnings("deprecation")
/**
 * @Title: MQ.WorkQueues.Worker.java 
 * @Package MQ.WorkQueues 
 * @Description: TODO(MQ消息发送到topic的交换机上) 
 * @Copyright: Copyright (c) 2017 YUANH All Rights Reserved
 * @author yuanh
 * @date 2017-5-10 下午3:48:59
 */
public class WorkerResponse {
	// private final static String EXCHANGE_NAME = "topic_logs";
	private final static String QUEUE_NAME = "hello_topic1";

	// private final static String QUEUE_NAME2 = "hello_topic1";

	public static void main(String[] argv) throws Exception {

		// 创建连接连接到MabbitMQ
		ConnectionFactory factory = new ConnectionFactory();
		// 设置MabbitMQ所在主机ip或者主机名
		factory.setHost("127.0.0.1");
		factory.setUsername("yuanh");
		factory.setPassword("yuanh");
		factory.setPort(5672);
		factory.setVirtualHost("y_yuanh");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		// 1声明队列、设置队列持久化
		boolean durable = true;
		channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
		QueueingConsumer consumer = new QueueingConsumer(channel);
		// 2消费者指定消费队列，打开应答机制, 注意false才是打开手动应对,true为自动应答
		boolean ack = false;
		channel.basicConsume(QUEUE_NAME, ack, consumer);
		// 3消费者设置最大服务转发消息数量, 公平转发
		int prefetchCount = 1;
		channel.basicQos(prefetchCount);
		try {
			while (true) {
				QueueingConsumer.Delivery delivery = consumer.nextDelivery();
				String message = new String(delivery.getBody());
				System.out.println(" 接收 '" + message + "'");
				try {
					doWork(message);
				} finally {
					System.out.println("结束");
					// 另外需要在每次处理完成一个消息后，手动发送一次应答(ack=false)。
					channel.basicAck(delivery.getEnvelope().getDeliveryTag(),
							false);
				}
			}
		} catch (Exception e) {
			channel.close();
			connection.close();
		}
	}

	private static void doWork(String task) throws InterruptedException {
		for (char ch : task.toCharArray()) {
			if (ch == '.') {
				Thread.sleep(100);
			} else {
				Thread.sleep(100);
			}

		}
	}
}
